package com.ljm.boot.kafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.messaging.Message;

import javax.annotation.Resource;
import java.util.Properties;

/**
 * @Description 配置全局的消息发送回调监听
 * @Author Dominick Li
 * @CreateTime 2024/1/16 15:33
 **/
@Slf4j
@Configuration
public class KafkaConfig {

    public static final String DEFALUT_TOPIC = "autoTopic";

    public static final String CONSUMER_LISTENER_ERROR_HANDLER_NAME ="consumerAwareListenerErrorHandler";

    @Resource
    ProducerFactory producerFactory;

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        KafkaTemplate<String, Object> kafkaTemplate = new KafkaTemplate<String, Object>(producerFactory);
        kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {
            @Override
            public void onSuccess(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata) {
                log.info("send susscess , data= {}", producerRecord.toString());
            }

            @Override
            public void onError(ProducerRecord<String, Object> producerRecord, Exception exception) {
                //当消息发送失败可以拿到消息存在缓存或数据中 定时重试发送
                log.error("send fail , error:{}", exception.getMessage());
            }
        });
        return kafkaTemplate;
    }

    /**
     * 消费异常监听器
     */
    @Bean(CONSUMER_LISTENER_ERROR_HANDLER_NAME)
    public ConsumerAwareListenerErrorHandler consumerAwareListenerErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {
            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
                log.error("consumer fail：{}" ,exception.getMessage());
                return null;
            }
        };
    }

    @Bean
    public AdminClient adminClient(@Value("${spring.kafka.bootstrap-servers}") String bootstrapServers) {
        Properties prop = new Properties();
        prop.put("bootstrap.servers", bootstrapServers);
        return AdminClient.create(prop);
    }

    @Bean
    public NewTopic newTopic() {
        return TopicBuilder.name(DEFALUT_TOPIC)
                .partitions(1)
                .replicas(1)
                .build();
    }
}

